package com.antstorm.statistics.service.mqtt;

import com.antstorm.statistics.config.MqttConfigProperty;
import com.antstorm.statistics.util.MacSignature;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;

@Component
@Data
@NoArgsConstructor
@AllArgsConstructor
public class MqttConsumer {

    @Autowired
    private MqttConfigProperty mqttConfigProperty;

    @Autowired
    private MqttConsumerHandler mqttConsumerHandler;

    private static MqttClient consumer;

    private static MqttConnectOptions ops;

    @PostConstruct
    private void initConsumer() throws MqttException, NoSuchAlgorithmException, InvalidKeyException {
        String passwd = MacSignature.macSignature(mqttConfigProperty.getGroupId(), mqttConfigProperty.getKeySecret());
        MemoryPersistence memoryPersistence = new MemoryPersistence();
        // 创建mqtt客服端
        consumer = new MqttClient(mqttConfigProperty.getBorker(), mqttConfigProperty.getClientId(), memoryPersistence);
        // 创建连接参数
        ops = new MqttConnectOptions();
        // 重新启动和重新连接时清空sesseion
        ops.setCleanSession(true);
        ops.setUserName(mqttConfigProperty.getUsername());
        ops.setPassword(passwd.toCharArray());
        // 设置超时时间 ,单位：秒
        ops.setConnectionTimeout(10);
        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
        ops.setKeepAliveInterval(20);
        // 设置自动重连
        ops.setAutomaticReconnect(true);
        // 设置回调函数
        consumer.setCallback(mqttConsumerHandler);
        // 订阅
        subscribe();
    }


    /**
     * 订阅
     */
    public void subscribe() throws MqttException {
        if(!consumer.isConnected()){
            consumer.connect(ops);
        }
        consumer.subscribe(mqttConfigProperty.getTopic(), mqttConfigProperty.getQos());
    }
}
